package com.northpool.operator.analysis.overlap;

import com.alibaba.fastjson.JSON;
import com.northpool.operator.annotation.InputParam;
import com.northpool.operator.annotation.InputParams;
import com.northpool.operator.annotation.OperatorInfo;
import com.northpool.operator.annotation.OutputParam;
import com.northpool.operator.annotation.OutputParams;
import com.northpool.operator.annotation.ParamEnum;
import com.northpool.operator.annotation.ParamEnums;
import com.northpool.operator.common.algorithm.OperatorTestBuild;
import com.northpool.operator.common.algorithm.TestParamBuilder;
import com.northpool.operator.common.database.DatabaseUtils;
import com.northpool.operator.common.stpclient.StpClientOperatorBase;
import com.northpool.operator.param.ParamTool;
import com.northpool.operator.param.custom.DatasetParam;
import com.northpool.operator.type.TypeEnum;
import com.northpool.resources.datasource.IDataSource;
import com.northpool.resources.datasource.PasswordCrypo;
import com.northpool.resources.datatable.ITable;
import com.northpool.spatial.grid.Constants;
import com.northpool.stpclient.bean.BaseResponse;
import com.northpool.stpclient.bean.overlay.OverlayAndMergeToDBPlan;
import com.northpool.stpclient.bean.tile.dataset.QueryDataSet;
import com.northpool.stpclient.bean.tile.dataset.SqlDataSet;
import com.northpool.stpclient.constants.OverlapConstants;
import com.northpool.stpclient.modules.geoprocess.overlay.IOverlayOperator;

@OperatorInfo(
        ref = "IntersectFlinkOperator",
        name = "相交",
        description = "叠加分析-相交算子",
        enums = {
                @ParamEnums(ref = "mode", value = {
                        @ParamEnum(ref = "A", name = "A", description = "数据集A的字段"),
                        @ParamEnum(ref = "B", name = "B", description = "数据集B的字段"),
                        @ParamEnum(ref = "ALL", name = "ALL", description = "数据集A和B字段的并集"),
                })
        }
)
@OutputParams({
        @OutputParam(ref = "dataset", name = "结果数据集"),
        @OutputParam(ref = "dataSourceId", name = "结果数据源ID"),
        @OutputParam(ref = "tableName", name = "结果表名称"),
})
public class IntersectFlinkOperator extends StpClientOperatorBase {
    String dataSourceIdA;
    String tableNameA;
    String fieldsA;
    String sqlFilterA;
    String dataSourceIdB;
    String tableNameB;
    String fieldsB;
    String sqlFilterB;
    String targetDataSourceId;
    String targetTableName;
    Integer targetLevel;
    Boolean addArea;
    OverlapConstants.SAVE_TYPE saveType;
    OverlapConstants.SAVE_FIELD_TYPE saveFieldType;

    @InputParams({
            @InputParam(ref = "datasetA", name = "数据集A", desc = "数据集包含数据源ID和数据表名称的JSON描述，如: {\"dataSourceId\":\"xxx-xxx\",\"tableName\":\"table\"}", type = TypeEnum.DB, required = true),
            @InputParam(ref = "fieldsA", name = "数据字段A", desc = "数据字段列表，多个逗号隔开，如: f_id,dlmc,dlbm", type = TypeEnum.STRING, required = false),
            @InputParam(ref = "sqlFilterA", name = "SQL过滤条件A", desc = "SQL过滤条件, 如: dlmc = '旱地'", type = TypeEnum.STRING, required = false),
            @InputParam(ref = "datasetB", name = "数据集B", desc = "数据集包含数据源ID和数据表名称的JSON描述，如: {\"dataSourceId\":\"xxx-xxx\",\"tableName\":\"table\"}", type = TypeEnum.DB, required = true),
            @InputParam(ref = "fieldsB", name = "数据字段B", desc = "数据字段列表，多个逗号隔开，如: f_id,dlmc,dlbm", type = TypeEnum.STRING, required = false),
            @InputParam(ref = "sqlFilterB", name = "SQL过滤条件B", desc = "SQL过滤条件, 如: dlmc = '旱地'", type = TypeEnum.STRING, required = false),
            @InputParam(ref = "targetDataSourceId", name = "目标数据源ID", type = TypeEnum.DATASOURCE, required = true),
            @InputParam(ref = "targetTableName", name = "目标表名称", desc = "目标表名称", type = TypeEnum.STRING, required = true),
            @InputParam(ref = "targetLevel", name = "目标层级", type = TypeEnum.NUMBER, required = false, value = "12"),
//            @InputParam(ref = "saveType", name = "重叠保存类型",type = TypeEnum.STRING, required = false, value = "INTERSECTION", testValue = "INTERSECTION", desc = "重叠操作返回类型。有4种，A，B，ALL，INTERSECTION，默认是INTERSECTION。必须大写。其中，INTERSECTION表示两空间重叠部分，A表示空间1部分减去INTERSECTION，B表示空间2部分减去INTERSECTION，ALL表示A+B+INTERSECTION"),
            @InputParam(ref = "saveFieldType", name = "重叠保存字段类型", type = TypeEnum.STRING, required = false, value = "A", desc = "重叠操作返回字段类型，仅在saveType=INTERSECTION时生效。有3种，A，B，ALL，默认是ALL。必须大写。其中，A表示仅关注数据1字段属性，会合并数据1的同一要素；B表示仅关注数据2字段属性，会合并数据2的同一要素；ALL表示关注数据1、数据2字段属性。"),
            @InputParam(ref = "addArea", name = "是否计算面积", desc = "是否计算面积，返回面积统计字段，默认字段名_area_", type = TypeEnum.BOOL, required = false, value = "false"),
    })
    public IntersectFlinkOperator() {
        super();
    }

    public static void main(String[] args) throws Exception {
        OperatorTestBuild.execute(IntersectFlinkOperator.class,
                TestParamBuilder.aBuilder()
                        .param("datasetA", "{\"dataSourceId\":\"402881ac934fb41701934fc212af000b\",\"tableName\":\"public.dltb_2022_nmk_3857\"}")
                        .param("datasetB", "{\"dataSourceId\":\"402881ac934fb41701934fc212af000b\",\"tableName\":\"public.dltb_2023_nmk_3857\"}")
                        .param("targetDataSourceId", "402881ac934fb41701934fc212af000b")
                        .param("targetTableName", "result.dltb_nmk_相交结果")
                        .param("saveFieldType", "ALL")
                        .param("addArea", "true")
                        .build());
    }

    @Override
    public void execute(ParamTool params) throws Exception {
        // 输入参数
        DatasetParam datasetA = params.getDataset("datasetA");
        DatasetParam datasetB = params.getDataset("datasetB");

        this.dataSourceIdA = datasetA.getDataSourceId();
        this.tableNameA = datasetA.getTableName();
        this.fieldsA = params.getStr("fieldsA");
        this.sqlFilterA = params.getStr("sqlFilterA");
        this.dataSourceIdB = datasetB.getDataSourceId();
        this.tableNameB = datasetB.getTableName();
        this.fieldsB = params.getStr("fieldsB");
        this.sqlFilterB = params.getStr("sqlFilterB");
        this.targetLevel = params.getNumber("targetLevel").intValue();
        this.addArea = params.getBool("addArea");

        this.targetDataSourceId = params.getStr("targetDataSourceId");
        this.targetTableName = params.getStr("targetTableName");
//        this.saveType = OverlapConstants.SAVE_TYPE.valueOf(params.get("saveType"));
        this.saveType = OverlapConstants.SAVE_TYPE.INTERSECTION;
        this.saveFieldType = OverlapConstants.SAVE_FIELD_TYPE.valueOf(params.getStr("saveFieldType"));

        String targetDataset = DatabaseUtils.buildDatasetJson(this.targetDataSourceId, this.targetTableName);
        this.result("dataset", targetDataset);
        this.result("dataSourceId", this.targetDataSourceId);
        this.result("tableName", this.targetTableName);

        IOverlayOperator overlayOperator = this.client.overlayOperator();
        OverlayAndMergeToDBPlan plan = new OverlayAndMergeToDBPlan();
        QueryDataSet dataSetA = new QueryDataSet(new SqlDataSet(PasswordCrypo.encode(DatabaseUtils.getDataSourceJson(this.dataSourceIdA)), this.tableNameA, this.sqlFilterA, this.fieldsA));
        QueryDataSet dataSetB = new QueryDataSet(new SqlDataSet(PasswordCrypo.encode(DatabaseUtils.getDataSourceJson(this.dataSourceIdB)), this.tableNameB, this.sqlFilterB, this.fieldsB));

        plan.setTileDataSet1(dataSetA);
        plan.setTileDataSet2(dataSetB);
        plan.setSaveType(saveType);
        plan.setSaveFieldType(saveFieldType);
        plan.setTargetDataSourceStr(PasswordCrypo.encode(DatabaseUtils.getDataSourceJson(this.targetDataSourceId)));
        plan.setTargetTableName(this.targetTableName);
        plan.setTargetLevel(this.targetLevel);
        plan.setAddArea(this.addArea);
        String dataSourceJson = DatabaseUtils.getDataSourceJson(this.dataSourceIdA);
        IDataSource ds = IDataSource.fromJSON(dataSourceJson);
        ITable table = ds.getTable(this.tableNameA);
        Integer srid = table.spatialInfo().getSrid();
        Constants.GRID_UNIT gridUnit = Constants.GRID_UNIT.getGridUnitBySrid(srid);
        plan.setGridUnit(gridUnit);

        String requestBody = JSON.toJSONString(plan);
        log(requestBody);

        BaseResponse response = overlayOperator.startOverlay(plan);
        overlayOperator.onStart(response.getRequestId(), () -> {
            this.log("叠加开始");
        });
        overlayOperator.onSuccess(response.getRequestId(), () -> {
            this.log("叠加成功");
            successExit();
        });
        overlayOperator.onError(response.getRequestId(), () -> {
            this.error("叠加失败: {}", overlayOperator.log(response.getRequestId()));
            this.failedExit();
        });
        this.waiting(24);
    }
}
